#include <moying_comm_lib/request.h>
#include <moying_comm_lib/utility.h>
#include <thread>
#include <iostream>

namespace moying
{
namespace communication
{
// void dummy_status_func(MoyingCommuStatus status, const std::string &msg)
// {
  
// }

MoyingCommuRequest::MoyingCommuRequest(const std::string &url,
                                 const std::string &identity) : socket_(MoyingCommuBase::ctx_, zmq::socket_type::req)
{
  MoyingCommuBase::url_ = url;
  MoyingCommuBase::identity_ = identity;
  poll_timeout_ms_ = 5 * 1000;  //默认请求超时时间
}

MoyingCommuRequest::MoyingCommuRequest(const std::string &ip, const unsigned int port, const std::string &identity)
:MoyingCommuRequest(std::string("tcp://" + ip + ":" + std::to_string(port)), identity)
{
}

MoyingCommuRequest::~MoyingCommuRequest()
{
  close();
}

bool MoyingCommuRequest::run()
{

  status_call_ = std::bind(&MoyingCommuRequest::status_cb,this,std::placeholders::_1,std::placeholders::_2);
  if (is_started_ || is_stoped_)
    return false;

  try
  {
    if(heartbeat_timeout_ms_ > 0){
      socket_.setsockopt(ZMQ_HEARTBEAT_IVL, heartbeat_interval_ms_);
      socket_.setsockopt(ZMQ_HEARTBEAT_TIMEOUT, heartbeat_timeout_ms_);
    }
    socket_.setsockopt(ZMQ_RECONNECT_IVL, kReconnIVL);
    socket_.setsockopt(ZMQ_RECONNECT_IVL_MAX, kReconnMax);
    socket_.setsockopt(ZMQ_LINGER, kLingerTimeout);
    socket_.setsockopt(ZMQ_REQ_RELAXED, 1);//允许在收到应答前发送新的请求,发送和应答不是一一对应
    
 
    std::ostringstream o;
    o<<serial_++;
    std::string inproc_name("inproc://monitor-request");
    inproc_name += o.str();

    monitor_start_waiter_->set_signaled(false);
    createMonitor(socket_, inproc_name);
    if(!monitor_start_waiter_->signaled())  //防止notify后再wait
        monitor_start_waiter_->wait();

    socket_.connect(url_);

    createLoop();
  }
  catch (zmq::error_t e)
  {
    callLog(MoyingCommuLogLevels::FATAL, "MoyingCommuRequest::run() throw %s err_code:%d\n",
            e.what(), e.num());
    return false;
  }
  is_started_ = true;
  return true;
}

/**
 * @brief MoyingCommuRequest::restart  清除旧的连接，建立新的连接
 * @return
 */
bool MoyingCommuRequest::restart(){

    //释放旧资源
    is_restart_ = true;
    while (true)
    {
      wait_send_queue_.broadcast();
      SLEEP(10);
      if (is_monitor_exit_)
        break;
    }
    socket_.close();
    if(discard_packet_before_connect_){
        std::cout<<"\n=========clear send queue\n";
        clearSendQueue();
    }


    //创建新的socket
    try{
       socket_ = zmq::socket_t(MoyingCommuBase::ctx_,zmq::socket_type::req);
    }catch(zmq::error_t e){
        std::cout<<"create req socket failed: "<<e.what()<<std::endl;
        callLog(MoyingCommuLogLevels::FATAL,"create req socket failed: %s",e.what());
        is_restart_ = false;
        return false;
    }

    is_restart_ = false;
    is_started_ = false;
    bool ret = run();
    if(ret)
        is_monitor_exit_ = false;

    return ret;

}

bool MoyingCommuRequest::close()
{

  if (is_stoped_)
    return false;

  callLog(MoyingCommuLogLevels::TRACE, "%s","enter request mode socket close\n");
  is_stoped_ = true;
  auto start = std::chrono::system_clock::now();
  while (is_started_)
  {
    if (is_loop_exit_ && is_monitor_exit_)
      break;
    wait_send_queue_.broadcast();
    SLEEP(10);
  }

  socket_.close();
  clearSendQueue();
  is_started_ = false;
  callLog(MoyingCommuLogLevels::TRACE, "%s","leave request mode socket close\n");
  return true;
}

void MoyingCommuRequest::status_cb(MoyingCommuStatus status, const std::string &msg){
    if(status == MoyingCommuStatus::CONNECTED){
        is_connected_ = true;
        wait_send_queue_.broadcast();
        is_timeout_ = false;

        // if(is_timeout_){    //由超时引起的重新连接不再通知调用者
        //     is_timeout_ = false;
        //     return;
        // }
    }else if(status == MoyingCommuStatus::DISCONNECTED ){
        is_connected_ = false;
        // if(is_timeout_ && disconnect_after_timeout_){    //由超时重连引起的连接断开不再通知调用者
        //     disconnect_after_timeout_ = false;
        //     return;
        // }
    }else if(status == MoyingCommuStatus::TIMEOUT){
        is_timeout_ = true;
        // disconnect_after_timeout_ = true;
    }
    if(is_stoped_){ //对象將要释放时，忽略掉任何通知
        return;
    }
    if(user_define_status_cb_)
      user_define_status_cb_(status,msg);
}

/**
 * @brief MoyingCommuRequest::setStatusCall    设置通讯状态回调，更新连接标志
 * @param func
 * @return
 */
void MoyingCommuRequest::setStatusCall(StatusCall func){
    // 必须设置StatusCall用于更新_is_connected, 否则requestAndWait无法使用
    // if(func == nullptr){
    //     status_call_ = &dummy_status_func;
    // }
    user_define_status_cb_ = func;
}

void MoyingCommuRequest::setDiscardPacketBeforeConntect(bool param)
{
    discard_packet_before_connect_ = param;
}

/**
 * @brief clearSendQueue     清空发送队列
 * @return
 */
void MoyingCommuRequest::clearSendQueue()
{
  std::lock_guard<std::mutex> lk(mutex_send_queue_);
  while (!queue_send_.empty())
    queue_send_.pop();
}

/**
 * @brief printPackWrapper      打印通讯数据包
 * @param is_send               true:发送; false:接收
 * @param pack                  数据包
 * @return
 */
void MoyingCommuRequest::printPackWrapper(bool is_send, const std::string &content, bytes_ptr pack, int thread_id)
{
  if (print_pack_call)
  {
    if(proto_types_.find(content) == proto_types_.end())
    {
      // WARNING("Content "<<content<<" not found in registered proto types");
      return;
    }

    char* p = pack->data();
    int total_size  = GET_INT(p);
    int header_size = GET_INT(p);
    p = pack->data();
    bytes_ptr data = std::make_shared<bytes_vec>(p+header_size , p+total_size);
    OFFSET(p,8);
    int identity_len = GET_INT(p);
    std::string identity(p,identity_len);
    OFFSET(p,identity_len);
    int pack_id = GET_INT(p);
    long long timestamp = GET_LONGLONG(p);
    timestamp = ntohll(timestamp);

    moying_proto::ServicePack whole_pack;
    whole_pack.ParseFromArray(data->data(),data->size());
    std::string data_str;
    if(is_send)
    {
      proto_types_[content].first->ParseFromArray(whole_pack.request().data(),whole_pack.request().size());
      google::protobuf::TextFormat::PrintToString(*proto_types_[content].first, &data_str);
    }
    else
    {
      proto_types_[content].second->ParseFromArray(whole_pack.response().data(),whole_pack.response().size());
      google::protobuf::TextFormat::PrintToString(*proto_types_[content].second, &data_str);
    }
    
    // std::string fancy_msg = "\n\033[1m\033[30m[Moying]:\033[0m" + std::to_string(timestamp) 
    //             + (is_send ? " [Request: " : " [Response: " ) + identity + "-\033[35m" + content + "\033[0m-" + std::to_string(pack_id) + "]\n";
    std::string time_str = getDateTime(timestamp);
    std::string fancy_msg = "\n[Moying]:" + time_str 
                + (is_send ? " [Request: " : " [Response: " ) + identity + "-" + content + "-" + std::to_string(pack_id) + "]\n";
    print_pack_call(fancy_msg + data_str);
  }
}

/**
 * @brief invokeRecvCall        接收回调调用
 * @param data                  数据内容
 * @return
 */
void MoyingCommuRequest::invokeRecvCall(bytes_ptr data)
{
  if (recv_call_ != nullptr)
    recv_call_(nullptr, data, nullptr);
}

/**
 * @brief createLoop            创建主功能循环线程
 * @return
 */
void MoyingCommuRequest::createLoop()
{
  auto pkg = [&]() -> void {
    auto thread_id = getTid();
    callLog(MoyingCommuLogLevels::INFO, "[tid:%d] started request loop.\n", thread_id);

    std::vector<zmq_pollitem_t> poll_vec;
    poll_vec.push_back({socket_, 0, ZMQ_POLLIN, 0});

    // bool is_timeout = false;
    ReqData req_data("", nullptr, nullptr, true);
    is_loop_exit_ = false;

    while (!is_stoped_)
    {
      try
      {
        // if(!is_timeout)
        {
          // 从发送队列中取出一个业务数据包
          {
            std::unique_lock<std::mutex> lk(mutex_send_queue_);
            if (queue_send_.empty() || !is_connected_)
            {
              lk.unlock();
              wait_send_queue_.wait();
              continue;
            }
            req_data = queue_send_.front();
            // queue_send_.pop();
          }
          // 把业务数据包封装到 protobuf 数据里
          auto pack_send = encodeSendBuf(req_data.data_, seq_id_);
          ++seq_id_;

          // 发送 protobuf 数据, 根据需要打印内容
          zmq::message_t msg_send(pack_send->data(), pack_send->size());
          socket_.send(msg_send);
          if(req_data.func_)
            response_callbacks_[req_data.content_] = std::make_pair(req_data.func_, req_data.log_);
          if(req_data.log_)
            printPackWrapper(true, req_data.content_, pack_send, thread_id);
        }

        // 拉取应答数据
        zmq::poll(poll_vec, poll_timeout_ms_);
        if (poll_vec[0].revents & ZMQ_POLLIN)
        {
          zmq::message_t msg_recv;
          socket_.recv(&msg_recv);

          // 解析收取的 protobuf 数据包, 根据需要打印内容
          bytes_ptr pack_recv = std::make_shared<bytes_vec>((char*)msg_recv.data(),(char*)msg_recv.data()+msg_recv.size());
          bytes_ptr data_recv = decodeRecvBuf(pack_recv);
          moying_proto::ServicePack data = getProtobufMessage<moying_proto::ServicePack>(data_recv);
          std::string content = data.topic();
          if(response_callbacks_.find(content) != response_callbacks_.end())
          {
            if(response_callbacks_[content].second)
              printPackWrapper(false, content, pack_recv, thread_id);
            // 调用接收回调函数
            response_callbacks_[content].first->processService(data.request(), data.response());
            response_callbacks_.erase(content);
          }
          // is_timeout = false;
          std::unique_lock<std::mutex> lk(mutex_send_queue_);
          queue_send_.pop();
        }else
        {
          // 发送的数据应答超时, 通知调用方
          // is_timeout = true;
          if(is_connected_){  //如果连接状态下请求超时,就不再重复请求了,因为可能是服务端处理慢或者阻塞
            std::unique_lock<std::mutex> lk(mutex_send_queue_);
            queue_send_.pop();
          }
          // zmq::message_t msg_recv;
          // socket_.recv(&msg_recv,ZMQ_DONTWAIT);
          invokeStatusCall(MoyingCommuStatus::TIMEOUT, url_);
        }

      }
      catch (zmq::error_t e)
      {
        callLog(MoyingCommuLogLevels::FATAL, "[tid:%d] MoyingCommuRequest loop throw %s err_code:%d\n",
                thread_id, e.what(), e.num());
        if (e.num() == ETERM) // 上下文已终止
          break;
      }
    } // while
    callLog(MoyingCommuLogLevels::INFO, "[tid:%d] stoped request loop.\n", thread_id);

    is_loop_exit_ = true;


    // if(is_timeout && !is_stoped_){
    //     callLog(MoyingCommuLogLevels::DEBUG, "[tid:%d] reconnect after timeout.\n", thread_id);
    //     if(!restart())  //超时后，重新创建socket连接
    //          is_loop_exit_ = true;
    // }else
    //     is_loop_exit_ = true;


  };

  std::thread thr(pkg);
  thr.detach();
}
} // namespace communication
} // namespace moyingrobot
